package com.apps.sdses.flink141.sql;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author jiwei
 * @description
 * @date 2023/8/29 16:33
 */
public class Kafka2Doris {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        env.enableCheckpointing(10000);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //Source
//        tableEnv.executeSql("CREATE TABLE sourceTable (\n" +
//                "    user_id        BIGINT,\n" +
//                "    username        STRING,\n" +
//                "    city        STRING,\n" +
//                "    age        INT,\n" +
//                "    sex        INT,\n" +
//                "    phone   INT,\n" +
//                "    address   STRING,\n" +
//                "    register_time   TIMESTAMP(3)\n" +
//                ") WITH (\n" +
//                "  'connector' = 'datagen', \n" +
//                "  'number-of-rows' = '5', \n" +
////                "  'rows-per-second' = '1', \n" +
//                "  'fields.username.length' = '5', \n" +
//                "  'fields.city.length' = '10', \n" +
//                "  'fields.address.length' = '10', \n" +
//                "  'fields.user_id.min' = '1', \n" +
//                "  'fields.user_id.max' = '10', \n" +
//                "  'fields.phone.min' = '11', \n" +
//                "  'fields.phone.max' = '12', \n" +
//                "  'fields.age.min' = '0', \n" +
//                "  'fields.age.max' = '100'\n" +
//                ")");

                tableEnv.executeSql("create table sourceTable (\n" +
                        "id bigint, \n" +
                        "pdId string, \n" +
                        "ewsensorId bigint,\n" +
                        "taskTime timestamp(3),\n" +
                        "saveTime timestamp(3),\n" +
                        "realTime timestamp(3),\n" +
                        "mpNo string,\n" +
                        "equipNo string, \n" +
                        "equipTypeCode string,\n" +
                        "status int,\n" +
                        "eWSensorGswd double,\n" +
                        "eWSensorHswd double,\n" +
                        "eWSensorGsyl double,\n" +
                        "eWSensorHsyl double,\n" +
                        "alarmCode string,\n" +
                        "errorDetail string,\n" +
                        "extendField string\n" +
                        ") with (\n" +
                        "  'properties.bootstrap.servers' = '172.21.32.233:39092,172.21.32.234:39092,172.21.32.235:39092',\n" +
                        "  'connector' = 'kafka',\n" +
                        "  'format' = 'json',\n" +
                        "  'topic' = 'ODS_SC01_SecondNetworkSensor_drpt',\n" +
                        "  'json.ignore-parse-errors' = 'true',\n" +
                        "  'properties.group.id' = 'ODS_SC01_SecondNetworkSensor_drpt_20240814',\n" +
                        "  'scan.startup.mode' = 'latest-offset'\n" +
                        ")");



        //Sink
//        tableEnv.executeSql("CREATE TABLE sinkTable (" +
//                "   user_id BIGINT, \n" +
//                "   username STRING, \n" +
//                "   city STRING, \n" +
//                "   age INT, \n" +
//                "   sex INT, \n" +
//                "   phone INT, \n" +
//                "   address STRING, \n" +
//                "   register_time TIMESTAMP(3)\n" +
//                "   ) " +
//                "   WITH (\n" +
//                "  'connector' = 'doris',\n" +
//                "  'fenodes' = '192.168.102.158:38030',\n" +
//                "  'table.identifier' = 'db_datacube_doris_test.user_info_unique',\n" +
//                "  'username' = 'datacube_doris_test',\n" +
//                "  'password' = 'Nx8Tz_Y8v3x',\n" +
////                "  'sink.properties.format' = 'json',\n" +
//                "  'sink.label-prefix' = 'doris_label5'\n" +
////                "  'sink.buffer-count' = '1'\n" +
//                ")");

        tableEnv.executeSql("CREATE TABLE sinkTable WITH ('connector' = 'print')\n" +
                "LIKE sourceTable (EXCLUDING ALL)");

        //Insert
        tableEnv.executeSql("insert into sinkTable select * from sourceTable");



    }
}
